package com.etc.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.ArrayList;


/**
 * @author 杰哥
 * @Title: TransformBlacklist
 * @ProjectName scalaworldcount
 * @Description: TODO
 * @date 2019/8/2018:47
 */
public class TransformBlacklist {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("TransformBlacklist").setMaster("local[*]");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
        ArrayList<Tuple2<String, Boolean>> blacklist = new ArrayList<Tuple2<String, Boolean>>();
        final JavaPairRDD<String, Boolean> blacklistRDD = jsc.sparkContext().parallelizePairs(blacklist);
        JavaReceiverInputDStream<String> master = jsc.socketTextStream("master", 8888);
        JavaPairDStream<String, String> JPDS = master.mapToPair(new PairFunction<String, String, String>() {
            @Override
            public Tuple2<String, String> call(String s) throws Exception {
                return new Tuple2<String, String>(s.split(" ")[1],s);
            }
        });
        JavaDStream<String> transform = JPDS.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {

            @Override
            public JavaRDD<String> call(JavaPairRDD<String, String> string) throws Exception {
                JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> JavaPairRDD = string.leftOuterJoin(blacklistRDD);
                org.apache.spark.api.java.JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filter = JavaPairRDD.filter(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, Boolean>() {
                    @Override
                    public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {
                        if(tuple._2._2().isPresent()&&tuple._2._2.get()){
                            return false;
                        }
                        return true;
                    }
                });
                JavaRDD<String> map = filter.map(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, String>() {
                    @Override
                    public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> stringTuple2Tuple2) throws Exception {
                        return stringTuple2Tuple2._2._1;
                    }
                });
                return map;
            }
        });
        transform.print();
        jsc.start();
        try{
            jsc.awaitTermination();
        }catch (Exception e ){
            e.printStackTrace();
        }

        jsc.close();
    }
}
